/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月22日 下午1:48:53
*/
package cn.com.guage.flink.broadcast;

/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月22日 下午1:48:53
*/
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * broadcast广播变量
 *
 * 需求：
 * flink会从数据源中获取到用户的姓名
 * 最终需要把用户的姓名和年龄信息打印出来
 * 所以就需要在中间map处理的时候，就需要获取用户的年龄信息
 * 建议将用户的关系数据集使用广播变量进行处理
 * 注意：如果多个算子需要使用同一份数据集，那么需要在对应的多个算子后面分别注册广播变量
 */
public class BatchDemoBroadcast {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //1、准备广播变量的数据
        ArrayList<Tuple2<String, Integer>> broadData = new ArrayList<>();
        broadData.add(new Tuple2<>("python",18));
        broadData.add(new Tuple2<>("scala",20));
        broadData.add(new Tuple2<>("java",17));
        DataSource <Tuple2 <String, Integer>> dataBroad = env.fromCollection(broadData);

        //2、对需要广播的数据进行处理，将tuple2类型转换成hashMap类型
        DataSet<HashMap <String, Integer>> baseData = dataBroad.map(new MapFunction <Tuple2 <String, Integer>, HashMap <String, Integer>>() {
            /**
			 * 
			 */
			private static final long serialVersionUID = 6565617523831046764L;

			@Override
            public HashMap <String, Integer> map(Tuple2 <String, Integer> value) throws Exception {
                HashMap <String, Integer> res = new HashMap <>();
                res.put(value.f0, value.f1);
                return res;
            }
        });
        baseData.print();
        DataSet <String> mainData = env.fromElements("python", "java","java","kafka","scala","redis");

        DataSet <String> result = mainData.map(new RichMapFunction <String, String>() {
            /**
			 * 
			 */
			private static final long serialVersionUID = -9066224106535622575L;
			List <HashMap <String, Integer>> broadCastMap = new ArrayList <HashMap <String, Integer>>();
            HashMap <String, Integer> allMap = new HashMap <String, Integer>();

            /**
             * 这个方法只会执行一次
             * 可以在这里实现一些初始化的功能
             *
             * 所以，就可以在open方法中获取广播变量数据
             *
             */
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                //3:获取广播数据
                this.broadCastMap = getRuntimeContext().getBroadcastVariable("broadCastMapName");
                for (HashMap<String, Integer> map : broadCastMap) {
                    allMap.putAll(map);
                }

            }

            @Override
            public String map(String value) throws Exception {
                Integer age = allMap.get(value);
                return value + "," + age;
            }
        }).withBroadcastSet(baseData,"broadCastMapName");

        result.print();
    }
}

